package org.way.rocketMq.service;

import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
import org.way.rocketMq.bean.RocketMqProperties;
import org.way.rocketMq.dto.MessageDto;
import org.way.rocketMq.dto.MqConstant;

import javax.annotation.Resource;

@Slf4j
@Component
public class MqProducerService {

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    @Resource
    private RocketMqProperties mqProperties;

    public void sendMsg(String message) {
        MessageDto messageDto = new MessageDto(message);
        rocketMQTemplate.convertAndSend(mqProperties.getTopic(),messageDto);
    }

    /**
     * 发送同步消息（阻塞当前线程，等待broker响应发送结果，这样不太容易丢失消息）
     * （message也可以是对象，sendResult为返回的发送结果）
     * 同步消息常用于比较重要的消息发送，需要等待broker响应告知消息发送状态
     */
    public SendResult sendSyncMsg(String message) {
        MessageDto messageDto = new MessageDto(message);
        SendResult sendResult = rocketMQTemplate.syncSend(mqProperties.getTopic(), MessageBuilder.withPayload(messageDto).build());
        log.info("syncSendMsg result:{}", JSONUtil.toJsonStr(sendResult));
        return sendResult;
    }

    /**
     * 发送异步消息（通过线程池执行发送到broker的消息任务，执行完后回调：在SendCallback中可处理相关成功失败时的逻辑）
     * （适合对响应时间敏感的业务场景）
     * 常用于对想用时间敏感，需要快速返回的模块，我们会设置一个回调代码块去异步监听Borker的响应
     */
    public void senAsyncMsg(String message) {
        MessageDto messageDto = new MessageDto(message);
        rocketMQTemplate.asyncSend(mqProperties.getTopic(), MessageBuilder.withPayload(messageDto).build(), new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                // 发送成功处理
                log.info("发送消息成功:{}",sendResult.toString());
            }

            @Override
            public void onException(Throwable throwable) {
                // 发送异常处理
                log.info("发送消息失败:{}",throwable.getMessage());
            }
        });
    }

    /**
     * 发送延时消息（上面的发送同步消息，delayLevel的值就为0，因为不延时）
     * 在start版本中 延时消息一共分为18个等级分别为：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
     */
    public void sendDelayMsg(String message, int delayLevel,long timeout) {
        MessageDto messageDto = new MessageDto(message);
        rocketMQTemplate.syncSend(mqProperties.getTopic(), MessageBuilder.withPayload(messageDto).build(), timeout, delayLevel);
    }

    /**
     * 发送单向消息（只负责发送消息，不等待应答，不关心发送结果，如日志）
     * 主要用于对发送结果不敏感，不会影响业务的模块，无需监听broker响应，常用于日志发送等模块
     */
    public void sendOneWayMsg(String message) {
        MessageDto messageDto = new MessageDto(message);
        rocketMQTemplate.sendOneWay(mqProperties.getTopic(), MessageBuilder.withPayload(messageDto).build());
    }

    /**
     * 发送带tag的消息，直接在topic后面加上":tag"
     */
    public SendResult sendMsg(String message,String tag) {
        MessageDto messageDto = new MessageDto(message);
        log.info("sendMsg:{}", JSONUtil.toJsonStr(messageDto));
        return rocketMQTemplate.syncSend(mqProperties.getTopic() + MqConstant.split + tag, MessageBuilder.withPayload(messageDto).build());
    }

}
